package org.qy.star_server.core;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.qy.star_common.enums.ExchangeType;
import org.qy.star_common.message.Message;
import org.qy.star_common.message.MessageProperties;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * @author yinger
 * @description TODO
 * @date 2024/3/26 07:39
 **/
@Data
@Slf4j
public class User {
    @Data
    public static class UserGloballyUniqueIdentity {
        private String username;

        public UserGloballyUniqueIdentity(String username) {
            this.username = username;
        }

        @Override
        public boolean equals(Object object) {
            if (this == object) return true;
            if (object == null || getClass() != object.getClass()) return false;
            UserGloballyUniqueIdentity that = (UserGloballyUniqueIdentity) object;
            return Objects.equals(username, that.username);
        }

        @Override
        public int hashCode() {
            return Objects.hash(username);
        }
    }

    /*属性*/
    private String username;    // 用户名
    private String password;    // 密码
    private int deleted = 0;    // 删除标记

    /*构造*/
    public User(String username, String password) {
        this.username = username;
        this.password = password;
    }

    public User() {
    }


    @Override
    public boolean equals(Object object) {
        if (this == object) return true;
        if (object == null || getClass() != object.getClass()) return false;
        User user = (User) object;
        return Objects.equals(username, user.username);
    }

    @Override
    public int hashCode() {
        return Objects.hash(username);
    }

    /*op锁*/
    private final Object userLocker = new Object();
    private final Object virtualHostLocker = new Object();
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();

    /*核心api，如果操作成功，返回空串，失败则返回错误消息*/

    private boolean userAuthentication() {
        if (!StringUtils.hasLength(username) || !StringUtils.hasLength(password)) {
            return false;
        }
        String psd = BrokerServer.memory.selectUser(username).getPassword();
        if (!this.password.equals(psd)) {
            log.error("用户身份校验失败！username={}", username);
            return false;
        }
        return true;
    }

    private boolean virtualHostAuthentication(String virtualHostName) {
        // 用户认证
        if (!userAuthentication()) {
            return false;
        }

        // 虚拟机是否属于用户
        VirtualHost isExistsVirtualHost = BrokerServer.memory.selectVirtualHost(username, virtualHostName);
        if (isExistsVirtualHost == null) {
            log.error("虚拟机身份校验失败！username={}", username);
            return false;
        }
        return username.equals(isExistsVirtualHost.getUsername());
    }

    private boolean exchangeAuthentication(String virtualHostName, String exchangeName) {
        if (!virtualHostAuthentication(virtualHostName)) {
            return false;
        }
        Exchange exchange = BrokerServer.memory.selectExchange(username, virtualHostName, exchangeName);
        if (exchange == null) {
            log.error("交换机身份验证失败");
            return false;
        }
        return virtualHostName.equals(exchange.getVirtualHostName());
    }

    private boolean queueAuthentication(String virtualHostName, String queueName) {
        if (!virtualHostAuthentication(virtualHostName)) {
            return false;
        }
        Queue queue = BrokerServer.memory.selectQueue(username, virtualHostName, queueName);
        if (queue == null) {
            log.error("队列身份验证失败");
            return false;
        }
        return virtualHostName.equals(queue.getVirtualHostName());
    }

    public String userDeclare() {
        synchronized (userLocker) {
            try {
                User user = BrokerServer.memory.selectUser(username);
                if (user != null) {
                    return "用户已经存在，" + username;
                }
                if (!BrokerServer.disk.insertUser(username, password)) {
                    return "创建用户失败 ... " + username;
                }
                BrokerServer.memory.insertUser(username, password);
                return "";
            } catch (Exception e) {
                log.error("用户创建失败！username={}", username);
                return "用户创建异常 ... " + username;
            }
        }
    }

    public String userDelete() {
        synchronized (userLocker) {
            try {
                if (!userAuthentication()) {
                    return "ACCESS DENY" + username;
                }
                User user = BrokerServer.memory.selectUser(username);
                if (user == null) {
                    return "操作删除用户不存在" + username;
                }
                if (!BrokerServer.disk.deleteUser(username)) {
                    return "删除用户失败 ... " + username;
                }
                BrokerServer.memory.deleteUser(username);
                return "";
            } catch (Exception e) {
                log.error("用户删除失败！username={}", username);
                return "用户删除异常 ... " + username;
            }
        }
    }

    public String userUpdate(String newPassword) {
        synchronized (userLocker) {
            try {
                if (!userAuthentication()) {
                    return "ACCESS DENY" + username;
                }
                User user = BrokerServer.memory.selectUser(username);
                if (user == null) {
                    return "操作更新用户不存在，" + username;
                }
                if (!BrokerServer.disk.updateUser(username, newPassword)) {
                    return "用户更新失败 ... " + username;
                }
                BrokerServer.memory.updateUser(username, newPassword);
                return "";
            } catch (Exception e) {
                log.error("用户更新失败！username={}", username);
                return "用户更新异常 ... " + username;
            }
        }
    }

    public String virtualHostDeclare(String virtualHostName) {
        synchronized (virtualHostLocker) {
            try {
                if (!userAuthentication()) {
                    return "ACCESS DENY" + virtualHostName;
                }

                // 1. 查询用户是否已经拥有该虚拟机
                VirtualHost vh = BrokerServer.memory.selectVirtualHost(username, virtualHostName);
                if (vh != null) {
                    log.error("用户已经拥有该虚拟机！username={}, virtualHostName={}", username, virtualHostName);
                    return "用户已经拥有该虚拟机,请勿重复创建," + virtualHostName;
                }

                // vh == null
                // 2. 将虚拟机插入数据库 如果数据库插入失败，则内存不进行操作 同步数据
                if (!BrokerServer.disk.insertVirtualHost(username, virtualHostName)) {
                    log.error("虚拟机创建失败！username={}, virtualHostName={}", username, virtualHostName);
                    return "虚拟机创建失败 ... " + virtualHostName;
                }

                // 3. 将虚拟机插入内存
                BrokerServer.memory.insertVirtualHost(username, virtualHostName);
                return "";
            } catch (Exception e) {
                log.error("虚拟机创建失败！username={}, virtualHostName={}", username, virtualHostName);
                return "虚拟机创建异常 ... " + virtualHostName;
            }
        }
    }

    public String virtualHostDelete(String virtualHostName) {
        synchronized (virtualHostLocker) {
            try {
                if (!virtualHostAuthentication(virtualHostName)) {
                    return "ACCESS DENY" + virtualHostName;
                }
                if (!BrokerServer.disk.deleteVirtualHost(username, virtualHostName)) {
                    log.error("删除虚拟机失败！username={}, virtualHostName={}", username, virtualHostName);
                    return "删除虚拟机失败 ... " + virtualHostName;
                }

                BrokerServer.memory.deleteVirtualHost(username, virtualHostName);
                return "";
            } catch (Exception e) {
                log.error("删除虚拟机失败！username={}, virtualHostName={}", username, virtualHostName);
                return "删除虚拟机异常 ... " + virtualHostName;
            }
        }
    }

    public String exchangeDeclare(String virtualHostName, String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
        synchronized (exchangeLocker) {
            try {
                if (!virtualHostAuthentication(virtualHostName)) {
                    return "ACCESS DENY" + virtualHostName;
                }

                Exchange exchange = BrokerServer.memory.selectExchange(username, virtualHostName, exchangeName);
                if (exchange != null) {
                    log.error("用户已经拥有该交换机！username={}, virtualHostName={}, exchangeName={}", username, virtualHostName, exchangeName);
                    return "用户已经拥有该交换机,请勿重复创建," + exchangeName;
                }
                if (!BrokerServer.disk.insertExchange(username, virtualHostName, exchangeName, exchangeType, durable, autoDelete, arguments)) {
                    log.error("交换机创建失败！username={}, virtualHostName={}, exchangeName={}", username, virtualHostName, exchangeName);
                    return "交换机创建失败 ... " + exchangeName;
                }
                BrokerServer.memory.insertExchange(username, virtualHostName, exchangeName, exchangeType, durable, autoDelete);
                return "";
            } catch (Exception e) {
                log.error("创建交换机失败！username={}, exchangeName={}", username, exchangeName);
                return "创建交换机异常 ... " + exchangeName;
            }
        }
    }

    public String exchangeDelete(String virtualHostName, String exchangeName) {
        synchronized (exchangeLocker) {
            try {
                if (!exchangeAuthentication(virtualHostName, exchangeName)) {
                    return "ACCESS DENY" + virtualHostName;
                }

                if (!BrokerServer.disk.deleteExchange(username, virtualHostName, exchangeName)) {
                    log.error("删除交换机失败！username={}, virtualHostName={}, exchangeName={}", username, virtualHostName, exchangeName);
                    return "删除交换机失败 ... " + exchangeName;
                }
                BrokerServer.memory.deleteExchange(username, virtualHostName, exchangeName);
                return "";
            } catch (Exception e) {
                log.error("删除交换机失败！username={}, virtualHostName={}, exchangeName={}", username, virtualHostName, exchangeName);
                return "删除交换机异常 ... " + exchangeName;
            }
        }
    }

    public String queueDeclare(String virtualHostName, String queueName, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) {
        synchronized (queueLocker) {
            try {
                if (!virtualHostAuthentication(virtualHostName)) {
                    return "ACCESS DENY" + virtualHostName;
                }
                Queue queue = BrokerServer.memory.selectQueue(username, virtualHostName, queueName);
                if (queue != null) {
                    log.error("用户已经拥有该队列！username={}, virtualHostName={}, queueName={}", username, virtualHostName, queueName);
                    return "用户已经拥有该队列,请勿重复创建," + queueName;
                }
                if (!BrokerServer.disk.insertQueue(username, virtualHostName, queueName, durable, exclusive, autoDelete, arguments)) {
                    log.error("队列创建失败！username={}, virtualHostName={}, queueName={}", username, virtualHostName, queueName);
                    return "队列创建失败 ... " + queueName;
                }
                BrokerServer.memory.insertQueue(username, virtualHostName, queueName, durable, exclusive, autoDelete);
                return "";
            } catch (Exception e) {
                log.error("创建队列失败！username={}, virtualHostName={}, queueName={}", username, virtualHostName, queueName);
                return "创建队列异常 ... " + queueName;
            }
        }
    }

    public String queueDelete(String virtualHostName, String queueName) {
        synchronized (queueLocker) {
            try {
                if (!queueAuthentication(virtualHostName, queueName)) {
                    return "ACCESS DENY" + virtualHostName;
                }

                if (!BrokerServer.disk.deleteQueue(username, virtualHostName, queueName)) {
                    log.error("删除队列失败！username={}, virtualHostName={}, queueName={}", username, virtualHostName, queueName);
                    return "删除队列失败 ... " + queueName;
                }
                BrokerServer.memory.deleteQueue(username, virtualHostName, queueName);
                return "";
            } catch (Exception e) {
                log.error("删除队列失败！username={}, virtualHostName={}, queueName={}", username, virtualHostName, queueName);
                return "删除队列异常 ... " + queueName;
            }
        }
    }

    public String queueBind(String virtualHostName, String exchangeName, String queueName, String bindingKey) {
        synchronized (exchangeLocker) {
            synchronized (queueLocker) {
                try {
                    if (!exchangeAuthentication(virtualHostName, exchangeName)) {
                        return "ACCESS DENY" + virtualHostName;
                    }

                    Binding binding = BrokerServer.memory.selectBinding(username, virtualHostName, exchangeName, queueName);
                    if (binding != null) {
                        log.error("队列已经绑定！username={}, virtualHostName={}, exchangeName={}, queueName={}, bindingKey={}", username, virtualHostName, exchangeName, queueName, bindingKey);
                        return "队列已经绑定,请勿重复绑定," + queueName;
                    }

                    if (!BrokerServer.disk.insertBinding(username, virtualHostName, exchangeName, queueName, bindingKey)) {
                        log.error("绑定失败！username={}, virtualHostName={}, exchangeName={}, queueName={}, bindingKey={}", username, virtualHostName, exchangeName, queueName, bindingKey);
                        return "绑定失败 ... " + queueName;
                    }

                    BrokerServer.memory.insertBinding(username, virtualHostName, exchangeName, queueName, bindingKey);
                    return "";
                } catch (Exception e) {
                    log.error("绑定失败！username={}, virtualHostName={}, exchangeName={}, queueName={}, bindingKey={}", username, virtualHostName, exchangeName, queueName, bindingKey);
                    return "绑定异常 ... " + queueName;
                }
            }
        }
    }

    public String queueUnbind(String virtualHostName, String exchangeName, String queueName) {
        synchronized (exchangeLocker) {
            synchronized (queueLocker) {
                try {
                    if (!queueAuthentication(virtualHostName, queueName) || !exchangeAuthentication(virtualHostName, queueName)) {
                        return "ACCESS DENY" + virtualHostName;
                    }

                    Binding binding = BrokerServer.memory.selectBinding(username, virtualHostName, exchangeName, queueName);
                    if (binding == null) {
                        log.error("队列绑定不存在！！username={}, virtualHostName={}, exchangeName={}, queueName={}", username, virtualHostName, exchangeName, queueName);
                        return "队列绑定不存在,无法删除," + queueName;
                    }

                    if (!BrokerServer.disk.deleteBinding(username, virtualHostName, exchangeName, queueName)) {
                        log.error("删除绑定失败！username={}, virtualHostName={}, exchangeName={}, queueName={},", username, virtualHostName, exchangeName, queueName);
                        return "删除绑定失败 ... " + queueName;
                    }

                    BrokerServer.memory.deleteBinding(username, virtualHostName, exchangeName, queueName);
                    return "";
                } catch (Exception e) {
                    log.error("删除绑定失败！username={}, virtualHostName={}, exchangeName={}, queueName={}", username, virtualHostName, exchangeName, queueName);
                    return "删除绑定异常 ... " + queueName;
                }
            }
        }
    }

    public String basicPublish(String virtualHostName, String exchangeName, String routingKey, String messageId, int deliveryMode, byte[] body) {
        try {
            if (!exchangeAuthentication(virtualHostName, exchangeName)) {
                return "ACCESS DENY" + virtualHostName;
            }

            if (!Router.checkRoutingKeyValid(routingKey)) {
                log.error("发布消息，路由键不合法 ... " + routingKey);
                return "路由键不合法 ... " + routingKey;
            }
            Exchange exchange = BrokerServer.memory.selectExchange(username, virtualHostName, exchangeName);
            if (exchange.getExchangeType() == ExchangeType.DIRECT) {
                Queue targetQueue = BrokerServer.memory.selectQueue(username, virtualHostName, routingKey);
                if (targetQueue == null) {
                    log.error("消息发布失败！username={}, virtualHostName={}, exchangeName={}, routingKey={}, messageId={}, deliveryMode={}, messageBody={}", username, virtualHostName, exchangeName, routingKey, messageId, deliveryMode, body);
                    return "消息发布失败 ... " + messageId + ",直接交换机目标队列不存在";
                }
                // 构造消息
                Message message = Message.createMessageWithId(new MessageProperties(messageId, routingKey, deliveryMode), body, routingKey);
                sendMessage(targetQueue, message);
            } else {
                // TOPIC or FANOUT
                // 获取交换机下的所有绑定
                List<Binding> bindings = BrokerServer.memory.selectBindingsFromExchange(username, virtualHostName, exchangeName);
                for (Binding binding : bindings) {
                    Queue targetQueue = BrokerServer.memory.selectQueue(username, virtualHostName, binding.getQueueName());
                    if (targetQueue == null) {
                        log.error("消息发布失败！username={}, virtualHostName={}, exchangeName={}, routingKey={}, messageId={}, deliveryMode={}, messageBody={}", username, virtualHostName, exchangeName, routingKey, messageId, deliveryMode, body);
                        continue;
                    }

                    // 构造消息
                    Message message = Message.createMessageWithId(new MessageProperties(messageId, routingKey, deliveryMode), body, routingKey);
                    // 判定这个消息是否能转发给该队列
                    //    如果是 fanout ，所绑定的所有队列都要转发
                    //    如果是 topic  ，还需要判断 bindingKey 和 routingKey 是否匹配
                    if (!Router.route(exchange.getExchangeType(), binding, message)) {
                        continue;
                    }
                    sendMessage(targetQueue, message);
                }
            }
            return "";
        } catch (Exception e) {
            log.error("消息发布失败！username={}, virtualHostName={}, exchangeName={}, routingKey={}, messageId={}, deliveryMode={}, messageBody={}", username, virtualHostName, exchangeName, routingKey, messageId, deliveryMode, body);
            return "消息发布失败 ... " + messageId;
        }
    }

    public void sendMessage(Queue queue, Message message) {
        try {
            synchronized (queueLocker) {
                if (queue == null) {
                    log.error("队列不存在！queueName={}", queue.getQueueName());
                    return;
                }
                VirtualHost virtualHost = BrokerServer.memory.selectVirtualHost(queue.getUsername(), queue.getVirtualHostName());
                virtualHost.getTokenQueue().put(new Token(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName()));
                BrokerServer.disk.insertMessage(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName(), message);
                BrokerServer.memory.insertMessageToCenter(message);
                BrokerServer.memory.insertMessageToQueue(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName(), message);
                VirtualHost v = BrokerServer.memory.selectVirtualHost(queue.getUsername(), queue.getVirtualHostName());
                v.getTokenQueue().put(new Token(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName()));
            }
        } catch (Exception e) {
            log.error("消息发送失败！queueName={},messageId={}", queue.getQueueName(), message.getMessageId());
        }
    }

    public String basicConsume(String virtualHostName, String consumerTag, Token token, boolean autoAck) {
        try {
            if (!queueAuthentication(virtualHostName, token.getQueueName())) {
                return "ACCESS DENY" + token.getQueueName();
            }

            log.info("订阅消息成功，推送消息中 ... " + token.getQueueName());
            return "";
        } catch (Exception e) {
            log.error("消息消费失败！username={}, virtualHostName={}, consumerTag={}, autoAck={}, consumer={}", username, virtualHostName, consumerTag, autoAck);
            return "消费消息异常" + token.getQueueName();
        }
    }

    public String basicAck(String virtualHostName, String queueName, String messageId) {
        try {
            if (!queueAuthentication(virtualHostName, queueName)) {
                return "ACCESS DENY" + queueName;
            }
            Queue queue = BrokerServer.memory.selectQueue(username, virtualHostName, queueName);
            Message message = BrokerServer.memory.selectMessageFromWaitAckMessageQueue(username, virtualHostName, queueName, messageId);
            if (message == null) {
                log.error("消息不存在！username={}, virtualHostName={}, queueName={}, messageId={}", queue.getQueueName(), virtualHostName, queueName, messageId);
                return "消息不存在" + messageId;
            }

            BrokerServer.disk.deleteMessage(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName(), message);
            BrokerServer.memory.deleteMessageFromCenter(message);
            BrokerServer.memory.deleteMessageFromQueue(queue.getUsername(), virtualHostName, queueName, message);
            BrokerServer.memory.deleteMessageFromWaitAckMessageQueue(queue.getUsername(), virtualHostName, queueName, message);
            log.info("消息确认成功！username={}, virtualHostName={}, queueName={}, messageId={}", queue.getUsername(), virtualHostName, queueName, messageId);
            return "";
        } catch (Exception e) {
            log.error("消息确认失败！username={}, virtualHostName={}, queueName={}, messageId={}", username, virtualHostName, queueName, messageId);
            return "消息确认失败" + messageId;
        }
    }

}
